Bound cuDF MPP memory across concat and exchange - #31

Summary

• Add a configurable hard byte cap to CudfBatchConcat, including pre-flush admission checks, row-and-byte grouping in the shared utility, and zero-copy passthrough for a single oversized input.
• Enforce partial aggregation memory limits while consuming input and merge accumulated runs with bounded fan-in.
• Allow phase-separated MPP destinations to retain packed payloads in host memory, with independent host/device queue accounting and corrected broadcast backpressure accounting.

Motivation

Large-scale MPP queries exposed unbounded GPU-memory behavior in batch concatenation, keyed final aggregation, and shuffle hash joins. The existing concat byte threshold was only a performance flush target and did not bound each actual cudf::concatenate call. Inactive MPP destinations could also retain payloads and state on the GPU until their consumers started.

Design

batchConcatMaxBytes is a stability boundary independent of the existing performance target. CudfBatchConcat checks the next input before admitting it, and the shared concatenate utility applies the same byte bound as a final defense. Other utility callers retain their existing behavior unless they pass an explicit byte limit. The default hard cap is 1 GiB.

MPP host materialization retains uncompressed packed payloads in pageable host memory and uses a bounded reusable pinned buffer for D2H staging. This change does not add disk spill or host-payload compression.

Companion Spark-Gluten PR: https://github.com/NVIDIA/spark-gluten/pull/88

Validation

• ConfigTest: 5/5 passed
• CudfBatchConcatTest: 14/14 passed
• Focused AggregationTest: 3/3 passed
• UcxOutputQueueManagerTest: 21/21 passed
• Release libgluten.so and mpp-substrait-runner: linked successfully
• Targeted large-concat workload: every actual concatenate input remained below the configured 1 GiB cap
• Targeted string-heavy shuffle hash join: completed without CUDA error, OOM, executor failure, or disk spill
• Representative fast-path workloads: no confirmed performance regression across three runs

Full-scale keyed final aggregation without payload compression or disk spill requires reducing repeated intermediate state with bounded partition-local combine. This follow-up is tracked by NVIDIA/spark-gluten#86.